1).两个打印语句的结果是什么，对应的依赖是宽依赖还是窄依赖，为什么会是这个结果；
通过运行代码，查看job作业DAG图 可以看出
rdd1.join(rdd2) 运行的stage划分图，看到join过程划分出了一个新的stage，所以是宽依赖
rdd1.partitionBy(new HashPartitioner(3)).join(rdd2.partitionBy(new HashPartitioner(3))) 
运行的stage划分图，可以看到join与之上的分区操作在同一个stage中，
也就是说join没有shuffle发生，所以是窄依赖。
2).join操作何时是宽依赖，何时是窄依赖；
3).借助join 相关源码，回答以上问题。
根据join 相关源码
  def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))] = self.withScope {
    join(other, defaultPartitioner(self, other))
  }
rdd1.join(rdd2) 实际是调用的重载方法 rdd.join(other,partitioner)

  def join[W](other: RDD[(K, W)], partitioner: Partitioner): 
    RDD[(K, (V, W))] = self.withScope {
      this.cogroup(other, partitioner).flatMapValues( pair =>
        for (v <- pair._1.iterator; w <- pair._2.iterator) yield (v, w)
      )
    }
rdd1.join(rdd2,partitioner) 调用 cogroup 进行连接操作

  def cogroup[W](other: RDD[(K, W)], partitioner: Partitioner)
      : RDD[(K, (Iterable[V], Iterable[W]))] = self.withScope {
    if (partitioner.isInstanceOf[HashPartitioner] && keyClass.isArray) {
      throw new SparkException("HashPartitioner cannot partition array keys.")
    }
    val cg = new CoGroupedRDD[K](Seq(self, other), partitioner)
    cg.mapValues { case Array(vs, w1s) =>
      (vs.asInstanceOf[Iterable[V]], w1s.asInstanceOf[Iterable[W]])
    }
  }
cogroup  方法 初始化 CoGroupRDD 来进行操作

  override def getDependencies: Seq[Dependency[_]] = {
    rdds.map { rdd: RDD[_] =>
      if (rdd.partitioner == Some(part)) {
        logDebug("Adding one-to-one dependency with " + rdd)
        new OneToOneDependency(rdd)
      } else {
        logDebug("Adding shuffle dependency with " + rdd)
        new ShuffleDependency[K, Any, CoGroupCombiner](
          rdd.asInstanceOf[RDD[_ <: Product2[K, _]]], part, serializer)
      }
    }
  }
CoGroupRDD  实现了getDependencies 通过源码分析，发现 只有rdd1与rdd2 的分区相同时,
是oneToOne 窄依赖 否则会出现 shuffle 为宽依赖



